package com.dxysun.springboot.config;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;

import java.util.Properties;

/**
 * @author dongxiyan
 * @date 2020/3/15 17:15
 */
@Configuration
@EnableKafka
public class KafkaConsumerConfig
{

    @Value("${dxy.topic}")
    private String topic;

    @Value("${dxy.group.id}")
    private String group;

    private Properties getKafkaConfig()
    {
        Properties configProperties = new Properties();
        configProperties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        configProperties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        configProperties.setProperty("auto.offset.reset", "latest");
//        configProperties.setProperty("auto.offset.reset", "earliest");
        configProperties.setProperty("enable.auto.commit", "false");
        return configProperties;
    }

    @Bean("kafkaConsumerFactory")
    public ConsumerFactory kafkaConsumerFactoryBJ()
    {
        Properties configProperties = getKafkaConfig();
        configProperties.setProperty("bootstrap.servers", "dxy.dxysun.com:9092");
//        configProperties.setProperty("bootstrap.servers", "127.0.0.1:9092");

        configProperties.setProperty("group.id",group);
        return new DefaultKafkaConsumerFactory(configProperties);
    }

   /* @Bean("listenerContainer")
    public KafkaMessageListenerContainer listenerContainerBJ(
            @Qualifier("kafkaConsumerFactory") ConsumerFactory consumerFactory, ListenService listenService)
    {
        ContainerProperties containerProperties = new ContainerProperties(topic);
        containerProperties.setSyncCommits(true);
        containerProperties.setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL);
        containerProperties.setMessageListener(listenService);
        return new KafkaMessageListenerContainer(consumerFactory, containerProperties);
    }

    @Bean("listenerContainer1")
    public KafkaMessageListenerContainer listenerContainerBJ1(
            @Qualifier("kafkaConsumerFactory") ConsumerFactory consumerFactory, ListenAService listenService)
    {
        ContainerProperties containerProperties = new ContainerProperties(topic);
        containerProperties.setSyncCommits(true);
        containerProperties.setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL);
        containerProperties.setMessageListener(listenService);
        return new KafkaMessageListenerContainer(consumerFactory, containerProperties);
    }
*/
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory(
            ConsumerFactory consumerFactory)
    {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(consumerFactory);
        factory.setConcurrency(1);
        factory.getContainerProperties().setPollTimeout(3000);
        factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL);
        return factory;
    }
}
